package com.lxb.demo02

import org.apache.flink.api.common.functions.FilterFunction
import org.apache.flink.streaming.api.scala._;

/**
 * @author: albert
 * @date: 2021/5/20 15:53
 * @description:
 */
object FunctionDemo {

    def main(args:Array[String]):Unit = {

      val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

      // 获取数据
      val input = "D:\\project\\bigdata\\demoall\\flinkDemo\\src\\main\\resources\\sensor.txt"

      val dataStream: DataStream[Sensor] = env.readTextFile(input)
        .map(data => {
          val strings: Array[String] = data.split(",")
          Sensor(strings(0).trim, strings(1).trim.toLong, strings(2).trim.toDouble)
        })

      val filterData: DataStream[Sensor] = dataStream.filter(new MyFilter)

      filterData.print()

      env.execute()

    }
}

class MyFilter extends FilterFunction[Sensor]{
  override def filter(t: Sensor): Boolean = {
    t.name.startsWith("sensor_1")
  }
}
